package com.neusoft.neu24.nep;

import com.neusoft.neu24.nep.entity.SimpleMsg;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;


import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class Main {

    public static void main(String[] args) throws MQClientException, IOException {
//        new DefaultMQPullConsumer();
//        过时，不推荐使用 主动获得模式

//        生产者组，无所谓，但消费者组不行，非常关键，必须保证消费者组内的消费主题必须一致
//        消息向同一组消费者投递策略：广播模式和负载均衡
//        如果一个主题被多个组订阅，那么每个组给一次，组内参照前面
//        被动接收，执行立即性很强
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("default-group");
        consumer.setNamesrvAddr("localhost:9876");
//        订阅消费主题
//        "*" :消息过滤
        consumer.subscribe("broadcast-1", "*");
//        异步消费
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("  收到消息");
                System.out.println(list.get(0).toString());
                System.out.println("consumeConcurrentlyContext --> " + consumeConcurrentlyContext);
//                返回值: 决定消息是否出队列
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("consumer started");
//        Thread.sleep(86400000);
        System.in.read();
    }

    public static void main1(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("default-group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        producer.setSendMsgTimeout(1000);

        Message message = new Message("broadcast-1", "Hello world!".getBytes());

        SendResult result = producer.send(message);
        producer.shutdown();
        System.out.println(result.getSendStatus());
    }
}